package connector

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"
)

// ConsoleSinkConnector represents a console sink connector
type ConsoleSinkConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// format is the output format
	format string
}

// NewConsoleSinkConnector creates a new console sink connector
func NewConsoleSinkConnector() SinkConnector {
	return &ConsoleSinkConnector{
		metrics: make(map[string]interface{}),
		format:  "json",
	}
}

// Initialize initializes the connector
func (c *ConsoleSinkConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Get format
	if format, ok := config["format"].(string); ok {
		c.format = format
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *ConsoleSinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	c.started = true

	return nil
}

// Stop stops the connector
func (c *ConsoleSinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	c.started = false

	return nil
}

// Write writes records to the sink
func (c *ConsoleSinkConnector) Write(ctx context.Context, batch *RecordBatch) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if batch is valid
	if batch == nil || len(batch.Records) == 0 {
		return nil
	}

	// Write records
	for _, record := range batch.Records {
		// Format record
		var output string
		var err error
		switch c.format {
		case "json":
			output, err = c.formatJSON(record)
		case "text":
			output, err = c.formatText(record)
		default:
			output, err = c.formatJSON(record)
		}

		if err != nil {
			return fmt.Errorf("failed to format record: %w", err)
		}

		// Print record
		fmt.Println(output)
	}

	// Update metrics
	c.metrics["records_total"] = c.metrics["records_total"].(int) + len(batch.Records)
	c.metrics["last_write_time"] = time.Now().Unix()

	return nil
}

// Flush flushes any buffered records
func (c *ConsoleSinkConnector) Flush(ctx context.Context) error {
	// No buffering in console sink
	return nil
}

// GetMetrics gets connector metrics
func (c *ConsoleSinkConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// formatJSON formats a record as JSON
func (c *ConsoleSinkConnector) formatJSON(record *Record) (string, error) {
	// Parse value as JSON
	var value interface{}
	err := json.Unmarshal(record.Value, &value)
	if err != nil {
		// If not valid JSON, use as string
		value = string(record.Value)
	}

	// Create output
	output := map[string]interface{}{
		"key":       string(record.Key),
		"value":     value,
		"timestamp": record.Timestamp,
	}

	// Add headers if not empty
	if len(record.Headers) > 0 {
		output["headers"] = record.Headers
	}

	// Add metadata if not empty
	if len(record.Metadata) > 0 {
		output["metadata"] = record.Metadata
	}

	// Marshal to JSON
	data, err := json.MarshalIndent(output, "", "  ")
	if err != nil {
		return "", fmt.Errorf("failed to marshal record: %w", err)
	}

	return string(data), nil
}

// formatText formats a record as text
func (c *ConsoleSinkConnector) formatText(record *Record) (string, error) {
	// Format key
	key := string(record.Key)
	if key == "" {
		key = "<null>"
	}

	// Format value
	value := string(record.Value)
	if value == "" {
		value = "<null>"
	}

	// Format timestamp
	timestamp := record.Timestamp.Format(time.RFC3339)

	// Create output
	output := fmt.Sprintf("Key: %s, Value: %s, Timestamp: %s", key, value, timestamp)

	return output, nil
}
